/*
 * ===================================================================
 *  
 *      Filename: log_srv_protocol_udp.c
 *
 *   Description: This is the udp plugin,the encapulation packets
 *                are welcomed(pack small packets into a large packet)
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *   
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>
#include <linux/unistd.h>
#include <unistd.h>
#include "lib/list.h"
#include "lib/util.h"

#define  __USE_GNU
#include <sched.h>
#include <pthread.h>

#define WORKING_THREAD         3
#define WORKING_THREAD_MASK    3

thread_info_t udp_thread_pool[WORKING_THREAD];

int udp_sockfd = 0;
extern long recv_counter;
extern long recv_traffic;

inline int udp_buffer_enqueue(char *buff, thread_info_t *thread)
{
    int   ipos = 0;
    char* send_buff = NULL;
    /*for the tcp, we only have one queue*/
    
    pthread_mutex_lock(&counter_mutex);
    recv_counter ++;
    recv_traffic += *(short*)buff;
    pthread_mutex_unlock(&counter_mutex);
   
    ipos = enqueue(thread,buff);
    if(NORMAL_THRESHOLD == ipos)
    {
        //normal case, don't send the traffic
        return OK;
    }
    else if(HIGH_THRESHOLD == ipos)
    {
        //little space left,try to output
    }
    else if(URGENCY_THRESHOLD == ipos)
    {
        /*wait until the packet can be handled*/
        //while(URGENCY_THRESHOLD == enqueue(thread,buff));
        /*how to reduce the change of drop*/
        free(buff);
             
        printf("queue is full,send the packet anyway\n");
    }
    else
    {
        //not care until now
        printf("I don't know what's going on, drop the packet\n");
        free(buff);
        return FALSE;
    }

    while((send_buff = (char*)dequeue(thread)))
    {
        #ifdef DEBUG_LOG
        printf("recv buff2 %p\n",send_buff);
        #endif
        /*we process the packet suddenly*/
        main_framer_process_pkt(send_buff);
    }
    return OK;
}

void *udp_process_log(void * argv)
{
    thread_info_t *thread = (thread_info_t *)argv;
    int  recv_size,pkt_len;
    char *new_local_buff = NULL,*tmp_buff= NULL,*send_buff = NULL;
    char buff[MAX_LOG_SIZE];//define the local global buffer for the udp socket
    struct sockaddr_in remote_client;
    int    remote_client_len,ret;
    cpu_set_t mask;

    CPU_ZERO(&mask);
    CPU_SET(thread->cpu_affinity, &mask);
    ret = pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
    if (ret < 0)
        perror("sched_setaffinity");
    
    remote_client_len = sizeof(struct sockaddr);
    for(;;)
    {
        tmp_buff          = (char*)&buff;//get the buffer point
        /*first 2 used for the length, the last byte used for the new line*/
        recv_size = recvfrom(udp_sockfd,buff,MAX_LOG_SIZE,0,(struct sockaddr*)&remote_client,(socklen_t *)&remote_client_len);
        if(recv_size>0)
        {
            /*register the client address to the policy modules*/
            add_remote_client(remote_client.sin_addr.s_addr);
           
            #ifdef DEBUG_LOG
            printf("buff1 %p\n ",buff);
            for(ipos=0;ipos<recv_size;ipos++)
            {
                printf("%02x",buff[ipos]);
            }
            printf("\n");
            #endif   
            
            pkt_len = *((short*)buff);
            /*we construct the udp BUG packet, so we assume the pacekt contains the wholelittle udp packet,if not,crash!!!!!*/
            while((pkt_len <= recv_size) && recv_size)
            {
                new_local_buff = malloc(pkt_len);
                if(!new_local_buff)
                {
                    printf("no new memory for the udp packet\n");
                    break;/*no memory,skip this log BIG packet*/
                }
                else
                {
                    memcpy(new_local_buff,tmp_buff,pkt_len);
                    udp_buffer_enqueue(new_local_buff,thread);
                }
                tmp_buff  = tmp_buff  + pkt_len;
                recv_size = recv_size - pkt_len;
                pkt_len  = *((short *)tmp_buff);
            }
            continue;//try to get the next packet from the socket
        }
        else
        {
            if(-1 == recv_size)
            {
                if(EAGAIN == errno)
                {
                    #ifdef DEBUG_LOG
                    printf("recv time out, try to flush the buffer\n");
                    #endif
                    /*timeout,flush the output queue*/
                    //socket_timeout_trigger_flush_buffer = TRUE;
                }
                else
                    continue;
            }
            else
                continue; //skip the loop
        }

        /*socket time out,flush the thread private queue*/
        while((send_buff = (char*)dequeue(thread)))
        {
            #ifdef DEBUG_LOG
            printf("time out buff2 %p\n",send_buff);
            #endif
            /*we process the packet suddenly*/
            main_framer_process_pkt(send_buff);
        }
        continue;
   }
	return NULL;
}
int init_protocol_udp()
{
    unsigned int ipos;
    struct timeval tv;
    struct sockaddr_in serv_addr;

    udp_sockfd = socket(AF_INET,SOCK_DGRAM,0);
    if(udp_sockfd == -1)
    {
        perror("error to create the socket");
        exit(1);
    }

    /*enlarge the socket buffer*/
    ipos = 4 * 1024 *1024;
    setsockopt(udp_sockfd,SOL_SOCKET,SO_RCVBUF,(char*)&ipos,sizeof(ipos));

    /*set the address to reuse*/
    ipos = TRUE;
    setsockopt(udp_sockfd,SOL_SOCKET,SO_REUSEADDR,&ipos,sizeof(ipos));

    /*set the recv timeout value*/
    tv.tv_sec = 3;
    tv.tv_usec= 0;
    setsockopt(udp_sockfd,SOL_SOCKET,SO_RCVTIMEO,&tv,sizeof(tv));   

    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port   = htons(SPORT);
    //inet_aton("10.1.136.8",&serv_addr.sin_addr.s_addr);
    serv_addr.sin_addr.s_addr   = htonl(INADDR_ANY);//*((struct in_addr*)host->h_addr);
    bzero(&(serv_addr.sin_zero),8);

    bind(udp_sockfd,(struct sockaddr*)&serv_addr,sizeof(struct sockaddr));

    /*create the working thread,now we only create 3 thread,and the forth is the tcp thread*/
    for(ipos=0;ipos<WORKING_THREAD;ipos++)
    {
        udp_thread_pool[ipos].cpu_affinity = ipos;
        /*initialize the attribution*/
        pthread_attr_init(&(udp_thread_pool[ipos].attr));
        /*set the pthread affinity*/
        pthread_attr_setscope(&(udp_thread_pool[ipos].attr), PTHREAD_SCOPE_SYSTEM);
        /*init the queue*/
        init_queue(&(udp_thread_pool[ipos]),MAX_WORK_QUEUE);
        //init_queue(&(udp_thread_pool[ipos]),(128<<ipos));
        /*create the pthread*/
        pthread_create(&(udp_thread_pool[ipos].thread),&(udp_thread_pool[ipos].attr),udp_process_log,(void*)&udp_thread_pool[ipos]);
    }
    return OK;
}

int protocol_udp_join()
{
    unsigned int ipos;
    for(ipos=0;ipos<WORKING_THREAD;ipos++)
    {
        pthread_join(udp_thread_pool[ipos].thread,NULL);
    }
    return OK;
}

int finit_protocol_udp()
{
    int i = 0;

    for(i=0;i<WORKING_THREAD;i++)
    {
        printf("    stop the udp protocol thread %ld\n",udp_thread_pool[i].thread);
        pthread_cancel(udp_thread_pool[i].thread);
        /*destroy the queue*/
        printf("    destroy the pthread private queue NO.%d\n",i);
        finit_queue((thread_info_t *)&udp_thread_pool[i]);
    }
    printf("    success to unload the udp protocol modules\n");
    return OK;
}

